package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;

/**
 * @author Felix
 * @date 2023/1/3
 * 事实表动态分流
 */
public class BaseDBTableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    public BaseDBTableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    private HashMap<String,TableProcess> configMap = new HashMap<>();

    @Override
    public void open(Configuration parameters) throws Exception {
        //为了避免主流数据进来的时候，加载不到广播状态的配置信息，在open方法，先将配置信息提前加载一下
        Connection conn = null;
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            //注册驱动
            Class.forName("com.mysql.cj.jdbc.Driver");
            //建立连接
            conn = DriverManager.getConnection("jdbc:mysql://hadoop202:3306/gmall0725_config?" +
                "user=root&password=123456&useUnicode=true&" +
                "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
            //获取数据库操作对象
            ps = conn.prepareStatement("select * from gmall0725_config.table_process where sink_type = 'dwd'");
            //执行sql语句
            rs = ps.executeQuery();
            //处理结果集
            ResultSetMetaData metaData = rs.getMetaData();
            while (rs.next()){
                JSONObject jsonObj = new JSONObject();
                for (int i = 1; i <= metaData.getColumnCount(); i++) {
                    String columnName = metaData.getColumnName(i);
                    Object columnValue = rs.getObject(i);
                    jsonObj.put(columnName,columnValue);
                }
                TableProcess tableProcess = jsonObj.toJavaObject(TableProcess.class);
                String key = tableProcess.getSourceTable() + ":" + tableProcess.getSourceType();
                configMap.put(key,tableProcess);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //释放资源
            if(rs != null){
                rs.close();
            }
            if(ps != null){
                ps.close();
            }
            if(conn != null){
                conn.close();
            }
        }

    }

    //处理主流数据
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //获取操作的业务数据库表名
        String table = jsonObj.getString("table");
        //获取操作类型
        String type = jsonObj.getString("type");
        //拼接操作广播状态的key
        String key = table + ":" + type;
        //根据key 从广播状态中获取对应的配置信息
        TableProcess tableProcess = null;

        if((tableProcess = broadcastState.get(key)) != null||(tableProcess = configMap.get(key)) != null){
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");
            //过滤掉不需要传递的属性
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(dataJsonObj, sinkColumns);
            //补充输出目的地属性
            dataJsonObj.put("sink_table",tableProcess.getSinkTable());
            out.collect(dataJsonObj);
        }

    }

    //过滤掉不需要向下游传递的属性
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        List<String> columnList = Arrays.asList(sinkColumns.split(","));
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry->!columnList.contains(entry.getKey()));
    }

    //处理广播流数据
    // op：r {"before":null,"after":{"source_table":"base_province","source_type":"ALL","sink_table":"dim_base_province","sink_type":"dim","sink_columns":"id,name,region_id,area_code,iso_code,iso_3166_2","sink_pk":null,"sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1672730183578,"snapshot":"false","db":"gmall0725_config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1672730183578,"transaction":null}
    // op：c {"before":null,"after":{"source_table":"aa","source_type":"insert","sink_table":"aaa","sink_type":"dwd","sink_columns":"a,b","sink_pk":null,"sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1672730353000,"snapshot":"false","db":"gmall0725_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000013","pos":1324557,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1672730352820,"transaction":null}
    // op：u {"before":{"source_table":"aa","source_type":"insert","sink_table":"aaa","sink_type":"dwd","sink_columns":"a,b","sink_pk":null,"sink_extend":null},"after":{"source_table":"aa","source_type":"insert","sink_table":"aaa","sink_type":"dwd","sink_columns":"a,b,ccc","sink_pk":null,"sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1672730420000,"snapshot":"false","db":"gmall0725_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000013","pos":1324888,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1672730420049,"transaction":null}
    // op：d {"before":{"source_table":"aa","source_type":"insert","sink_table":"aaa","sink_type":"dwd","sink_columns":"a,b,ccc","sink_pk":null,"sink_extend":null},"after":null,"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1672730474000,"snapshot":"false","db":"gmall0725_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000013","pos":1325252,"row":0,"thread":null,"query":null},"op":"d","ts_ms":1672730474498,"transaction":null}
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        // System.out.println(jsonStr);
        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //将json字符串转换为json对象
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        String op = jsonObj.getString("op");
        if("d".equals(op)){
            //如果对配置表做的是删除操作，将对应的配置信息 从广播状态中删除掉
            TableProcess before = jsonObj.getObject("before", TableProcess.class);
            String sinkType = before.getSinkType();
            if("dwd".equals(sinkType)){
                String sourceTable = before.getSourceTable();
                String sourceType = before.getSourceType();
                String key = sourceTable + ":" + sourceType;
                broadcastState.remove(key);
                configMap.remove(key);
            }
        }else{
            //如果对配置表做的是删除外的其它操作，将对应的配置信息放到广播状态中
            TableProcess after = jsonObj.getObject("after", TableProcess.class);
            String sinkType = after.getSinkType();
            if("dwd".equals(sinkType)){
                String sourceTable = after.getSourceTable();
                String sourceType = after.getSourceType();
                String key = sourceTable + ":" + sourceType;
                broadcastState.put(key,after);
                configMap.put(key,after);
            }
        }
    }
}
